package com.mouse.hadoop.explorer.proxy;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;

/**
 * Created by cwx183898 on 2017/7/13.
 */
@Service
public class HdfsProxy {
    private Configuration config = new Configuration();
    @Value("${hdfs.defaultFS}")
    private String uri;

    public FileSystem getFileSystem() {
        try {
            FileSystem fs = FileSystem.get(URI.create(uri), config);
            return fs;
        } catch (Exception e) {
            return null;
        }
    }

    public boolean createDirectory(String path, String name) {
        try {
            if (path == null) {
                path = "/";
            }
            if (!path.endsWith("/")) {
                path = path + "/";
            }
            String fullPath = path + name;
            FileSystem fs = getFileSystem();
            if (fs.exists(new Path(fullPath))) {
                return false;
            }
            return fs.mkdirs(new Path(fullPath));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean delete(String path) {
        try {
            if (path == null || path.trim().length() == 0 || path.equalsIgnoreCase(uri) || path.equals("/")) {
                return false;
            }
            FileSystem fs = getFileSystem();
            Path target = new Path(path);
            if (fs.exists(target)) {
                return fs.delete(target, true);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public FileStatus getFileStatus(String path) {
        try {
            if (path == null || path.trim().length() == 0) {
                path = "/";
            }
            FileSystem fs = getFileSystem();
            return fs.getFileStatus(new Path(path));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public FileStatus[] getChildren(String path) {
        try {
            if (path == null || path.trim().length() == 0) {
                path = "/";
            }
            FileSystem fs = getFileSystem();
            if (fs != null) {
                return fs.listStatus(new Path(path));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public boolean saveFile(String path, String content) {
        FSDataOutputStream os = null;
        try {
            if (path == null || path.trim().length() == 0) {
                return false;
            }
            FileSystem fs = getFileSystem();
            os = fs.create(new Path(path), true);
            content = content.replaceAll("\n", "\r\n");
            os.write(content.getBytes());
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                //flush()刷新后数据不一定到了管道，hflush()刷新后可以保证数据都已经到达管道，hflush()=flush()+sync()
                os.hflush();
                os.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return false;
    }

    public boolean rename(String path, String name) {
        FSDataOutputStream os = null;
        try {
            if (path == null || path.trim().length() == 0) {
                return false;
            }
            FileSystem fs = getFileSystem();
            Path src = new Path(path);

            Path dest = new Path(path.substring(0, path.lastIndexOf("/") + 1) + name);
            fs.rename(src, dest);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean readFile(String path, OutputStream out) {
        try {
            if (path == null || path.trim().length() == 0) {
                return false;
            }
            Path inputPath = new Path(path);
            Configuration conf = new Configuration();
            FileSystem fs = getFileSystem();
            CompressionCodecFactory factory = new CompressionCodecFactory(conf);
            CompressionCodec codec = factory.getCodec(inputPath);
            //判断是否存在压缩编码，如果存在则表示是压缩文件，读取时先解压，否则为普通文件直接读取
            if (codec == null) {
                InputStream is = fs.open(inputPath);
                IOUtils.copyBytes(is, out, 1024, true);
            } else {
                InputStream is = codec.createInputStream(fs.open(inputPath));
                IOUtils.copyBytes(is, out, 1024, true);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }
}